package com.kqkj.gmall.canal.util;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class KafkaSender {
    public static KafkaProducer<String,String> kafkaProducer = null;

    public static KafkaProducer<String,String> createKafkaProducer(){
        Properties properties = new Properties();
        properties.put("bootstrap.servers","kqkj101:9092,kqkj102:9092,kqkj103:9092");
        properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String,String> producer = null;
        try {
            producer = new KafkaProducer<>(properties);
        } catch (Exception e){
            e.printStackTrace();
        }
        return producer;
    }

    public static void send(String topic,String msg){
        if(kafkaProducer == null){
            kafkaProducer = createKafkaProducer();
        }
        kafkaProducer.send(new ProducerRecord<String,String>(topic,msg));
    }
}
